package com.ds.lens.flink.monitor.metricsjob;

import com.ds.lens.flink.monitor.common.LensElasticsearchSink;
import com.ds.lens.flink.monitor.MonitorJobConstants;
import com.ds.lens.flink.monitor.MonitorJobContext;
import com.ds.lens.flink.monitor.common.CommonDeserializationSchema;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.http.HttpHost;

import java.util.List;
import java.util.Map;
import java.util.Properties;

public class LensMetricsJob {

    public static void main(String[] args) throws Exception {
        Properties properties = MonitorJobContext.getInstance().getKafkaProperties();
        if (properties == null) {
            System.err.println("No kafka properties found");
            return;
        }
        Properties kafkaProps = new Properties();
        kafkaProps.putAll(properties);
        kafkaProps.setProperty(MonitorJobConstants.KAFKA_CONSUMER_GROUP_CONFIG_FIELD, LensMetricsJob.class.getName());
        String metricsTopic = MonitorJobContext.getInstance().getMetricsTopic();
        Map<String, String> esConfig = MonitorJobContext.getInstance().getEsConfig();
        List<HttpHost> transportAddresses = MonitorJobContext.getInstance().getEsTransportAddresses();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(MonitorJobContext.getInstance().getMetricsJobParallelism());
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        DataStreamSource<Metrics> metricsDataStreamSource = env.addSource(new FlinkKafkaConsumer<>(metricsTopic,
                new CommonDeserializationSchema<>(Metrics.class), kafkaProps));

        //ES
        metricsDataStreamSource.process(new MetricsProcessFunction())
                .addSink(new LensElasticsearchSink<>(transportAddresses, new MetricsSinkFunction(), esConfig).build());

        //MYSQL
        metricsDataStreamSource.keyBy(new MetricsTagsKeySelector())
                .window(TumblingProcessingTimeWindows.of(Time.minutes(1)))
                .process(new MetricsTagsProcessFunction());

        env.execute(LensMetricsJob.class.getSimpleName());
    }


}
